package com.liuguang.rabbitmq.handler;

import com.alibaba.fastjson.JSON;
import com.liuguang.api.redis.RedisService;
import com.liuguang.common.constans.CommonConstans;
import com.liuguang.common.enums.CommonResultEnum;
import com.liuguang.common.pojo.CommonResult;
import com.liuguang.rabbitmq.pojo.MyCorrelationData;
import org.apache.dubbo.config.annotation.DubboReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: MyResendMessageHandler
 * @Description： 定时任务实现：provider 向 broker 发送失败消息
 * @Author: wanghr
 * Date: 11:01 AM 2020/11/6
 * Version: 1.0
 */
@Component
public class MyResendMessageHandler implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(MyResendMessageHandler.class);

    @DubboReference
    private RedisService redisService;

    //创建定时线程池
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);

    private RabbitTemplate rabbitTemplate;

    public MyResendMessageHandler(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        scheduledExecutorService.scheduleWithFixedDelay(new ReSendTask(rabbitTemplate), 10, 10, TimeUnit.SECONDS);
    }

    class ReSendTask implements Runnable {
        private RabbitTemplate rabbitTemplate;

        public ReSendTask(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
        }

        @Override
        public void run() {
            //在redis 获取发送失败消息
            logger.info("-----------------------进入定时任务");
            CommonResult<List<String>> result = redisService.getList(CommonConstans.FAIL_MESSAGE_LIST_KEY);
            //当获取不到时候返回 fail
            if (CommonResultEnum.SUCCESS.getCode().equals(result.getCode())) {
                List<String> failMessageList = result.getData();
                if (!CollectionUtils.isEmpty(failMessageList)) {
                    logger.info("-------------失败重发消息数量：{}", failMessageList.size());
                    for (String failMessage : failMessageList) {
                        //再次发送消息
                        MyCorrelationData correlationData = JSON.parseObject(failMessage, MyCorrelationData.class);
                        logger.info("correlationData:{}", correlationData);
                        String id = correlationData.getId() + "--重发";
                        correlationData.setId(id);
                        try {
                            rabbitTemplate.convertAndSend(correlationData.getExchangeName(), correlationData.getRoutingKey(),
                                    correlationData.getMqData(), correlationData);
                        } catch (AmqpException e) {
                            logger.info("重发再次异常");
                            redisService.saveToList(CommonConstans.FAIL_MESSAGE_LIST_KEY, failMessage);
                            continue;
                        }
                    }
                    //删除已经处理消息，这时候不能删除key，因为可能后面还有失败的消息添加到list。
                    redisService.deleteListByInteger(CommonConstans.FAIL_MESSAGE_LIST_KEY, failMessageList.size());
                }else {
                    logger.info("---------------没有发送失败的消息");
                }

            }
        }
    }

}
